package jk.fox.node.node;

import jk.fox.node.NodeStatusListener;
import jk.fox.node.rule.IRule;
import jk.fox.common.RpcDict;
import jk.fox.common.exception.runtime.RpcNoAvailableServerException;
import jk.fox.engine.RpcNodeDefinition;
import jk.fox.registry.discovery.DiscoveryEvent;
import jk.fox.registry.discovery.DiscoveryListener;
import lombok.Getter;
import lombok.Setter;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Auther cuichao
 * @Description: RpcNode
 * @Date: create in 2020/4/17 22:35
 *
 */
@Getter
@Setter
public class RpcNode implements DiscoveryListener<DiscoveryEvent> ,NodeStatusListener {

    private static Logger logger = Logger.getLogger(RpcNode.class);

    private String domain;
    private RpcNodeDefinition definition;
    private IRule rule;
    private boolean enableSameSource;
    private volatile List<String> uriList = Collections.synchronizedList(new ArrayList<>());
    private volatile ConcurrentMap<String,NodeServer> serverPool = new ConcurrentHashMap<>();
    private Lock lock = new ReentrantLock();


    public RpcNode(String domain, IRule rule) {
        this.domain = domain;
        this.rule = rule;
    }


    public RpcNode(RpcNodeDefinition definition, IRule rule) {
        this.definition = definition;
        this.domain = definition.getId();
        this.rule = rule;
    }


    public boolean hasSameSource(){
        return enableSameSource && definition.isProvider() && definition.isConsumer();
    }

    /**
     * 并发导致的数组越界
     * 1.解决方法一: 判断轮询到的数据下标是否小于 urlList的长度或者数组越界
     * 符合条件轮询再次查找
     * 2.解决方法二: 在RpcNode中替换ServerList对象对 List<String> uriList = node.getUriList();
     * 得到的对象引用不会进行干扰 对算法层面透明化
     */
    public String choose(){
        String rule = this.rule.rule(this);
        if(rule == null){
            throw new RpcNoAvailableServerException(String.format("没有可用的Rpc服务: %s",domain));
        }
        return rule;
    }

    @Override
    public void listen(DiscoveryEvent event) {
        switch (event.getType()){
            case INIT_NODE:
                addNode(event.getData());
                break;
            case ADD_NODE:
                addNode(event.getData());
                break;
            case UPDATE_NODE:
                updateNode(event.getData());
                break;
            case REMOVE_NODE:
                removeNode(event.getData());
                break;
            default:
                break;
        }
    }

    @Override
    public void listen(String address, RpcDict.NodeStatus status, Exception e) {
        if(!uriList.contains(address)){
            return;
        }
        NodeServer server = serverPool.get(address);
        //server为null基本不会发生
        if(server == null ){
            server = createServer(address);
        }
        /**
         * 标记为markdown后必须由注册中心恢复并重建server
         */
        if(server.getStatus() != RpcDict.NodeStatus.MARKDOWN){
            logger.debug("server status has changed the server is "+server.getAddress()+" status: "+status);
            server.setStatus(status);
        }
    }

    public NodeServer getServer(String node){
        return serverPool.get(node);
    }


    public NodeServer createServer(String node){
        NodeServer server = new NodeServer(node);
        serverPool.put(node,server);
        return server;
    }


    public void addNode(List<String> nodes){
        lock.lock();
        try {
            List<String> newList = new ArrayList<>();

            newList.addAll(uriList);
            newList.addAll(nodes);
            uriList = newList;
            for (String node : nodes) {
                createServer(node);
            }
        }finally {
            lock.unlock();
        }
    }


    public void addNode(String node){
        lock.lock();
        try {
            List<String> newList = new ArrayList<>();
            newList.addAll(uriList);
            newList.add(node);
            uriList = newList;
            createServer(node);
        }finally {
            lock.unlock();
        }
    }


    public void updateNode(String node){
        lock.lock();
        try {
            List<String> newList = new ArrayList<>();
            newList.addAll(uriList);
            newList.remove(node);
            newList.add(node);
            uriList = newList;
        }finally {
            lock.unlock();
        }
    }


    public void removeNode(String node){
        lock.lock();
        try {
            List<String> newList = new ArrayList<>();
            newList.addAll(uriList);
            newList.remove(node);
            uriList = newList;
            NodeServer server = serverPool.get(node);
            server.setStatus(RpcDict.NodeStatus.MARKDOWN);
        }finally {
            lock.unlock();
        }
    }


    @Override
    public String getId() {
        return domain;
    }


    @Override
    public void unRegistry(String key) {

    }

    public String getDomain() {
        return domain;
    }

    public List<String> getUriList() {
        return Collections.unmodifiableList(uriList);
    }


    public void enableSameSource(boolean enableSameSource) {
        this.enableSameSource = enableSameSource;
    }
}
